#include <boost/scoped_array.hpp>
#include "Network.h"
#include "utils/logger.h"
#include "OpConsume.h"
#include "SerializableOutput.h"

using namespace std;
using namespace internal;

OpConsume::OpConsume(const ConsumeOperation& operation) {
	source = createOp(operation.source());
}

int OpConsume::doit() {
	LOG("Sending data to worker 0 (on thread %d).", network->getThreadId());

	SerializableOutput buffer(source->getOutput());

  int sourceRows;
	while ((sourceRows = source->doit()) != -1) {
		buffer.append(source->getOutput(), sourceRows);

		if (buffer.full()) {
			network->send(0, buffer.serializedData(), buffer.serializedLength(), CONSUME_MID_FLAG);
			LOG("OpConsume sending %d bytes.", buffer.serializedLength());
			buffer.clear();
		}
	}

	network->send(0, buffer.serializedData(), buffer.serializedLength(), CONSUME_END_FLAG);
	LOG("OpConsume sending last %d bytes.", buffer.serializedLength());

	return -1;
}
